"""
Case Type   : GSC功能模块
Case Name   : mot外表gsc缓存功能验证
Create At   : 2024/01/25
Owner       : @peilinqian
Description :
    1. 关闭增量检查点开关;打开gsc开关、motserver开关;
    2. 重启数据库生效;
    3. 创建外表，并插入1条数据，并查询;
    4. 查看syscache占用情况;
    5. 增加外表索引;
    6. 查看syscache占用情况;
    7. 删除表;
    8. 查看syscache占用情况;
    9. 循环进行外表删除创建查询操作;
    10. 并发循环查询外表;
Expect      :
    1. 关闭增量检查点开关;打开gsc开关、motserver开关; 修改成功
    2. 重启数据库生效; 重启成功
    3. 创建外表，并插入1条数据，并查询; 操作成功
    4. 查看syscache占用情况; 缓存信息正确
    5. 增加外表索引; 操作成功
    6. 查看syscache占用情况; 缓存信息正确
    7. 删除表; 操作成功
    8. 查看syscache占用情况; 表缓存信息占用记录为空
    9. 循环进行外表删除创建查询操作; 功能正常
    10. 并发循环查询外表; 查询结果会间断出现无表报错或者查询正常
History     :
    Modified by peilinqian at 2025-3-4:研发代码变更，增加motserver开关配置
"""
import os
import re
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node


class GscTestCase(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)}:start----')
        self.log.info("----this is setup----")
        self.pri_dbuser = Node(node='PrimaryDbUser')
        self.pri_root = Node(node='PrimaryRoot')
        self.tb_name = "tb_gsc0064"
        self.select_thread_num = 10
        self.com = Common()
        self.constant = Constant()
        self.pri_sh = CommonSH('PrimaryDbUser')
        step_txt = '----step0:查询enable_incremental_checkpoint初始值;----'
        self.log.info(step_txt)
        self.init_para1 = self.com.show_param("enable_incremental_checkpoint")
        step_txt = '----step0:查看enable_global_syscache初始配置值;----'
        self.log.info(step_txt)
        self.init_para2 = self.com.show_param("enable_global_syscache")
        self.init_para3 = self.com.show_param("enable_mot_server")

        self.create_sql = f"drop foreign table if exists {self.tb_name};" \
            f"create foreign table {self.tb_name} " \
            f"(id int not null, aa varchar(10) ,id2 int);" \
            f"insert into {self.tb_name} values(11, '1a', 11);" \
            f"select * from {self.tb_name};"
        self.select_sql = f"select * from {self.tb_name};"

    def test_main(self):
        step_txt = '----step1:关闭增量检查点开关;打开gsc开关、motserver开关; ' \
                   'expect:修改成功----'
        self.log.info(step_txt)
        msg = self.pri_sh.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'enable_incremental_checkpoint=off')
        self.assertTrue(msg, '执行失败:' + step_txt)
        msg = self.pri_sh.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f"enable_global_syscache= on")
        self.assertTrue(msg, '执行失败:' + step_txt)
        msg = self.pri_sh.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f"enable_mot_server= on")
        self.assertTrue(msg, '执行失败:' + step_txt)

        step_txt = '----step2:重启数据库生效; expect:重启成功----'
        self.log.info(step_txt)
        restart = self.pri_sh.restart_db_cluster()
        self.assertTrue(restart, '执行失败:' + step_txt)

        step_txt = '----step3:创建外表，并插入1条数据，并查询; expect:操作成功----'
        self.log.info(step_txt)

        result = self.pri_sh.execut_db_sql(self.create_sql)
        self.log.info(result)
        self.assertIn(self.constant.DROP_FOREIGN_SUCCESS_MSG, result,
                      "执行失败" + step_txt)
        self.assertIn(self.constant.CREATE_FOREIGN_SUCCESS_MSG, result,
                      "执行失败" + step_txt)
        self.assertIn('INSERT 0 1', result, "执行失败" + step_txt)
        self.assertIn('1 row', result, "执行失败" + step_txt)

        step_txt = '----step4:查看syscache占用情况; expect:缓存信息正确----'
        self.log.info(step_txt)
        check_sql = f"select * from gs_gsc_table_detail() where " \
            f"relname like '{self.tb_name}%'; "
        check_result = self.pri_sh.execut_db_sql(check_sql)
        self.log.info(check_result)
        self.assertIn("'id','aa','id2'", check_result, '执行失败:' + step_txt)
        self.assertIn("1 row", check_result, '执行失败:' + step_txt)

        step_txt = '----step5:增加外表索引; expect:操作成功----'
        self.log.info(step_txt)
        drop_sql = f"create index on {self.tb_name}(id);" \
            f"select * from {self.tb_name};"
        result = self.pri_sh.execut_db_sql(drop_sql)
        self.log.info(result)
        self.assertIn(self.constant.CREATE_INDEX_SUCCESS, result,
                      "执行失败" + step_txt)

        step_txt = '----step6:查看syscache占用情况; expect:缓存信息正确----'
        self.log.info(step_txt)
        check_sql = f"select * from gs_gsc_table_detail() where " \
            f"relname like '{self.tb_name}%'; "
        check_result = self.pri_sh.execut_db_sql(check_sql)
        self.log.info(check_result)
        self.assertIn("'id','aa','id2'", check_result, '执行失败:' + step_txt)
        self.assertIn("2 row", check_result, '执行失败:' + step_txt)

        step_txt = '----step7:删除表; expect:操作成功----'
        self.log.info(step_txt)
        drop_sql = f"drop foreign table if exists {self.tb_name};"
        result = self.pri_sh.execut_db_sql(drop_sql)
        self.log.info(result)
        self.assertIn(self.constant.DROP_FOREIGN_SUCCESS_MSG, result,
                      "执行失败" + step_txt)

        step_txt = '----step8:查看syscache占用情况; expect:表缓存信息占用记录为空----'
        self.log.info(step_txt)
        check_result = self.pri_sh.execut_db_sql(check_sql)
        self.log.info(check_result)
        self.assertIn("0 rows", check_result, '执行失败:' + step_txt)

        step9_txt = '----step9:循环进行外表删除创建查询操作; expect:功能正常----'
        self.log.info(step9_txt)
        exec_thread = ComThread(self.pri_sh.cycle_exec_sql,
                                args=(self.create_sql, 120, False,
                                      None, True))
        exec_thread.setDaemon(True)
        exec_thread.start()
        time.sleep(0.3)

        step10_txt = '----step10:并发循环查询外表;' \
                     'expect:查询结果会间断出现无表报错或者查询正常----'
        self.log.info(step10_txt)
        connect_thread = []
        connect_result = []
        for i in range(self.select_thread_num):
            connect_thread.append(ComThread(self.pri_sh.cycle_exec_sql, args=(
                self.select_sql, 120, False, None, True)))
            connect_thread[i].setDaemon(True)
            connect_thread[i].start()

        self.log.info('----获取步骤9执行结果----')
        exec_thread.join(600)
        exec_result = exec_thread.get_result()
        pattern = r'DROP FOREIGN TABLE\s+.*CREATE FOREIGN TABLE' \
                  r'\s+INSERT 0 1\s+.*\(1 row\).*?'
        for i in set(exec_result):
            if not re.search(pattern, i, re.S | re.M):
                self.log.info(i)
            self.assertIsNotNone(re.search(pattern, i, re.S | re.M),
                                 '执行失败:' + step9_txt)

        self.log.info('----获取步骤10执行结果----')
        pattern = r'ERROR:  relation ".*" does not exist on dn_6001'
        for i in range(self.select_thread_num):
            connect_thread[i].join(600)
            connect_result.extend(connect_thread[i].get_result())
        self.log.info('执行总次数' + str(len(connect_result)))
        set_msg = set(connect_result)
        self.log.info(set_msg)
        self.assertLessEqual(len(set_msg), 3, '执行失败:' + step10_txt)
        if len(set_msg) == 1:
            self.assertEqual(str(set_msg).count('id | aa | id2'), 1,
                             '执行失败:' + step10_txt)
        if len(set_msg) == 2:
            if str(set_msg).count('id | aa | id2') < 2:
                self.assertIsNotNone(
                    re.search(pattern, str(set_msg), re.S | re.M),
                    '执行失败:' + step10_txt)
            else:
                self.assertEqual(str(set_msg).count('id | aa | id2'),
                                 2, '执行失败:' + step10_txt)
        if len(set_msg) == 3:
            self.assertEqual(str(set_msg).count('id | aa | id2'), 2,
                             '执行失败:' + step10_txt)
            self.assertIsNotNone(
                re.search(pattern, str(set_msg), re.S | re.M),
                '执行失败:' + step10_txt)

    def tearDown(self):
        self.log.info('----this is tearDown----')
        step11_txt = '----清理mot外表; expect:清理成功----'
        self.log.info(step11_txt)
        drop_cmd = f"drop foreign table if exists {self.tb_name};checkpoint;"
        drop_result = self.pri_sh.execut_db_sql(drop_cmd)
        self.log.info(drop_result)

        step12_txt = '----还原增量检查点开关、gsc开关; expect:修改成功----'
        self.log.info(step12_txt)
        msg1 = self.pri_sh.execute_gsguc('set',
                                         self.constant.GSGUC_SUCCESS_MSG,
                                         f'enable_incremental_checkpoint'
                                         f'={self.init_para1}')
        msg2 = self.pri_sh.execute_gsguc('set',
                                         self.constant.GSGUC_SUCCESS_MSG,
                                         f"enable_global_syscache "
                                         f"={self.init_para2}")
        msg3 = self.pri_sh.execute_gsguc('set',
                                         self.constant.GSGUC_SUCCESS_MSG,
                                         f"enable_mot_server "
                                         f"={self.init_para3}")

        step13_txt = '----重启数据库生效; expect:重启成功----'
        self.log.info(step13_txt)
        restart = self.pri_sh.restart_db_cluster()

        step_txt = '----teardown断言----'
        self.log.info(f'----{os.path.basename(__file__)}:end----')
        self.log.info(step_txt)
        self.assertIn(self.constant.DROP_FOREIGN_SUCCESS_MSG, drop_result,
                      "执行失败" + step11_txt)
        self.assertIn(self.constant.CHECKPOINT_SUCCESS_MSG, drop_result,
                      "执行失败" + step11_txt)
        self.assertTrue(msg1, '执行失败:' + step12_txt)
        self.assertTrue(msg2, '执行失败:' + step12_txt)
        self.assertTrue(msg3, '执行失败:' + step12_txt)
        self.assertTrue(restart, '执行失败:' + step13_txt)
